package com.shujia.mllib

import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo06ImagePredict {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo06ImagePredict")
      .master("local[*]")
      .config("spark.sql.shuffle.partitions", "8")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 1、加载需要预测的数据 并进行同样的数据特征工程处理
    val sourceDF: DataFrame = spark
      .read
      .format("image")
      .load("Spark/data/mllib/data/images")

    val proDF: DataFrame = sourceDF
      .select($"image.origin" as "filePath", $"image.data" as "data")
      .as[(String, Array[Byte])]
      .map {
        case (filePath: String, data: Array[Byte]) =>
          val fileName: String = filePath.split("/").last
          val iArr: Array[Double] = data.map(i => {
            if (i.toInt < 0) {
              1.0
            } else {
              0.0
            }
          })
          Tuple2(fileName, Vectors.dense(iArr).toSparse)
      }
      .toDF("fileName", "features")

    // 2、加载模型
    val logisticRegressionModel: LogisticRegressionModel = LogisticRegressionModel.load("Spark/data/mllib/image")

    // 3、预测结果
    val resDF: DataFrame = logisticRegressionModel.transform(proDF)

    resDF.show()


  }

}
